package com.fuyun.microservice.connect;

import com.fuyun.microservice.codec.msg.BroadRequestMessage;
import com.fuyun.microservice.codec.msg.Header;
import com.fuyun.microservice.connect.pojo.ClientEndpoint;
import io.netty.channel.Channel;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RMap;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.redisson.api.listener.MessageListener;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
@Slf4j
public class BroadMessageAdaptor {
    private final  RMap<String, RetryableBroadRequestMessage> retryMap ;
    private final SessionManager sessionManager;
    private final RTopic flowMessageTopic;

    public BroadMessageAdaptor(SessionManager sessionManager, RedissonClient redissonClient) {
        this.sessionManager = sessionManager;
        flowMessageTopic = redissonClient.getTopic("flowMessage");
        onFlowMessage(flowMessageTopic);
        retryMap=redissonClient.getMapCache("gate-message-retries");
    }

    /**
     * 发送广播消息，消息需要流转到其它实例
     *
     * @param message
     */
    public void broad(BroadRequestMessage message) {
        Header header = message.getHeader();
        for (ClientEndpoint clientEndpoint : sessionManager.getChannels()) {
            if (clientEndpoint.getTypes().contains(message.getType())) {
                if (SessionManager.getHostName().equals(clientEndpoint.getServerHostName()) == false) {
                    log.debug("client not login this instance:{},should notify other instance:{}", SessionManager.getHostName(),
                            clientEndpoint.getServerHostName());
                    flowMessage(clientEndpoint.getClientId(), message);
                } else {
                    Channel channel = sessionManager.getChannel(clientEndpoint.getClientId());
                    channel.writeAndFlush(message);
                    if (message.getShouldAck() == 1) {
                        retryMap.put(keyed(header.getSequenceNo(), clientEndpoint.getClientId()),
                                RetryableBroadRequestMessage.of(message, clientEndpoint.getClientId()));
                    }
                }
            }
        }
    }


    /**
     * 指定目标发送
     *
     * @param message
     * @param clientId
     */
    public boolean send2Client(int clientId,BroadRequestMessage message) {
        Channel channel = sessionManager.getChannel(clientId);
        if (channel == null) {
            flowMessage(clientId,message);
            return true;
        }

        channel.writeAndFlush(message);
        return true;
    }

    /**
     * 流转到其他实例发送
     *
     * @param message
     * @param clientId
     */
    protected void flowMessage(int clientId, BroadRequestMessage message) {
        flowMessageTopic.publish(FlowBroadRequestMesage.of(message,clientId));
    }

    protected void  onFlowMessage(final RTopic flowMessageTopic){
        flowMessageTopic.addListener(FlowBroadRequestMesage.class, (charSequence, msg) -> {
            Channel channel=sessionManager.getChannel(msg.clientId);
            if(channel==null){
                return;
            }
            channel.writeAndFlush(msg.broadRequestMessage);
            if (msg.broadRequestMessage.getShouldAck() == 1) {
                retryMap.put(keyed(msg.broadRequestMessage.getHeader().getSequenceNo(), msg.clientId),
                        RetryableBroadRequestMessage.of(msg.broadRequestMessage, msg.clientId));
            }
        });
    }

    /**
     * 消息确认
     *
     * @param ch
     * @param sequenceNo
     */
    public void acknowledge(Channel ch, long sequenceNo) {
        ClientEndpoint session = sessionManager.getSession(ch);
        if (session != null) {
            retryMap.remove(keyed(sequenceNo, session.getClientId()));
        }
    }

    //todo:没有确认的消息会间隔10秒重发
    @Scheduled(cron = "0/10 * * * * ?")
    public void reSent() {
        //todo: add redislock
        for (Map.Entry<String, RetryableBroadRequestMessage> entry : retryMap.entrySet()) {
            RetryableBroadRequestMessage retryable = entry.getValue();
            if (retryable.retryTimes > 5) {
                log.debug("stop retry over than 5 times,message:{}", retryable.broadRequestMessage);
                continue;
            }

            send2Client(retryable.clientId,retryable.broadRequestMessage);
            retryable.retryTimes++;
        }
    }


    private String keyed(long sequenceNo, int clientId) {
        return String.format("%d-%d", sequenceNo, clientId);
    }


    @Data
    private static class FlowBroadRequestMesage{
        private int clientId;

        private BroadRequestMessage broadRequestMessage;

        public static RetryableBroadRequestMessage of(BroadRequestMessage message, int clientId) {
            RetryableBroadRequestMessage retryableBroadRequestMessage = new RetryableBroadRequestMessage();
            retryableBroadRequestMessage.broadRequestMessage = message;
            retryableBroadRequestMessage.clientId = clientId;
            return retryableBroadRequestMessage;
        }
    }

    @Data
    private static class RetryableBroadRequestMessage {
        private int retryTimes;
        private int clientId;

        private BroadRequestMessage broadRequestMessage;

        public static RetryableBroadRequestMessage of(BroadRequestMessage message, int clientId) {
            RetryableBroadRequestMessage retryableBroadRequestMessage = new RetryableBroadRequestMessage();
            retryableBroadRequestMessage.broadRequestMessage = message;
            retryableBroadRequestMessage.clientId = clientId;
            return retryableBroadRequestMessage;
        }
    }
}
